package com.learning.db.newfunction;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.springframework.beans.BeanUtils;
import java.text.SimpleDateFormat;
import java.io.IOException;
import java.math.BigDecimal;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Objects;

/**
 * ClassName: CusJob
 * Description:
 * Date: 2018/4/4 9:46 【需求编号】
 *
 * @author Sam Sho
 * @version V1.0.0
 */
public class CusJob extends Configured implements Tool {

    /**
     * 计算往期数据 Mapper
     */
    static class CulOldMapper extends Mapper<LongWritable, Text, Text, CusBean> {

        /**
         * 个人编号,作为一行数据的key
         */
        private Text aac999 = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 获取每行数据的值
            String lineValue = value.toString().trim();
            // 进行分割
            String[] arr = lineValue.split("\\|");
            aac999.set(arr[0]);
            CusBean dataBean = new CusBean();
            dataBean.loadDate(arr);
            dataBean.setFlag(0);
            context.write(aac999, dataBean);
        }
    }

    /**
     * 本期计算Mapper
     */
    static class CulMapper extends Mapper<LongWritable, Text, Text, CusBean> {

        /**
         * 个人编号,作为一行数据的key
         */
        private Text aac999 = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 获取每行数据的值
            String lineValue = value.toString().trim();
            // 进行分割
            String[] arr = lineValue.split("\\|");
            aac999.set(arr[0]);
            CusBean dataBean = new CusBean();
            dataBean.loadDate(arr);
            dataBean.setFlag(1);
            context.write(aac999, dataBean);
        }
    }

    /**
     * 计算Reduce
     */
    static class CulReduce extends Reducer<Text, CusBean, CusBean, NullWritable> {

        /**
         * 自负费用
         * rate1~2-普通参保人员支付比例
         */
        double[][] rate1 = {{0, 6000, 0, 800},
                {6000, 10000, 0.3, 800},
                {10000, 20000, 0.4, 2000}};

        /**
         * 普通人员-自负加合规费用比例
         */
        double[][] rate2 = {{20000, 50000, 0.5, 0},
                {50000, 100000, 0.6, 15000},
                {100000, 200000, 0.7, 45000},
                {200000, Double.MAX_VALUE, 0.8, 115000}};

        /**
         * 自负费用
         * 实时救助人员支付比例
         */
        double[][] rate3 = {{0.0000, 3000.0000, 0.0000, 800.0000},
                {3000.0000, 6000.0000, 0.3500, 800.0000},
                {6000.0000, 10000.0000, 0.3500, 1850.0000},
                {10000.0000, 20000.0000, 0.4500, 3250.0000}};

        /**
         * 实时救助人员-自负加合规费用比例
         */
        double[][] rate4 = {{20000.0000, 50000.0000, 0.5500, 0.0000},
                {50000.0000, 100000.0000, 0.6500, 16500.0000},
                {100000.0000, 200000.0000, 0.7500, 49000.0000},
                {200000.0000, Double.MAX_VALUE, 0.8500, 124000.0000}};

        @Override
        protected void reduce(Text key, Iterable<CusBean> values, Context context) throws IOException, InterruptedException {
            ArrayList<CusBean> oldDatas = new ArrayList<>();
            ArrayList<CusBean> thisDatas = new ArrayList<>();

            // 区分往期并且合并计算需要数据
            BigDecimal previousSumAkc254 = new BigDecimal("0");
            BigDecimal previousSumComplianceOwnExpense = new BigDecimal("0");
//            BigDecimal previousSumAkc254AddComplianceOwnExpense = new BigDecimal("0");

            // 往期累计自负赔付
            BigDecimal previousSumAkc254Compensate = new BigDecimal("0");
            // 往期累计自负加合规赔付
            BigDecimal previousSumAkc254AddCOECompensate = new BigDecimal("0");

            for (CusBean value : values) {
                CusBean dataBean = new CusBean();
                BeanUtils.copyProperties(value, dataBean);
                int flag = dataBean.getFlag();
                // 先计算往期
                if (0 == flag) {
                    // 自负往期累计
                    BigDecimal bigDecimal = new BigDecimal(dataBean.getAkc254());
                    previousSumAkc254 = previousSumAkc254.add(bigDecimal);
                    // 合规自费 往期累计
                    BigDecimal bigDecimal2 = new BigDecimal(dataBean.getComplianceOwnExpense());
                    previousSumComplianceOwnExpense = previousSumComplianceOwnExpense.add(bigDecimal2);
                    // 自负+合规自费 往期累计
//                    BigDecimal bigDecimal3 = new BigDecimal(dataBean.getAkc254AddComplianceOwnExpense());
//                    previousSumAkc254AddComplianceOwnExpense = previousSumAkc254AddComplianceOwnExpense.add(bigDecimal3);

                    oldDatas.add(dataBean);
                } else {
                    thisDatas.add(dataBean);
                }
            }

            // 累计自负赔付
            BigDecimal sumAkc254Compensate = new BigDecimal("0");
            // 累计自负加合规赔付
            BigDecimal sumAkc254AddCOECompensate = new BigDecimal("0");

            Collections.sort(thisDatas);

            for (CusBean thisData : thisDatas) {

                // 本次累计+ 往期累计
                BigDecimal sumAkc254 = previousSumAkc254.add(new BigDecimal(thisData.getSumAkc254()));
                BigDecimal sumComplianceOwnExpense = previousSumComplianceOwnExpense.add(new BigDecimal(thisData.getSumComplianceOwnExpense()));
                BigDecimal sumAkc254AddComplianceOwnExpense = sumAkc254.add(sumComplianceOwnExpense);

                if (Objects.equals("0", thisData.getCkc005())) {
                    // 自负金额-赔付计算
                    if (sumAkc254.doubleValue() > 20000) {
                        sumAkc254Compensate = new BigDecimal("6000");
                    } else if (sumAkc254.doubleValue() < 6000) {
                        sumAkc254Compensate = new BigDecimal("0");
                    } else {
                        for (double[] doubles : rate1) {
                            if (sumAkc254.doubleValue() > doubles[0] && sumAkc254.doubleValue() <= doubles[1]) {
//                                sumAkc254Compensate = (sumAkc254 - doubles[0]) * doubles[2] + doubles[3];
                                sumAkc254Compensate = (sumAkc254.subtract(new BigDecimal(doubles[0]))).multiply(new BigDecimal(doubles[2])).add(new BigDecimal(doubles[3]));
                                break;
                            }
                        }
                    }

                    // 自负+ 计算后的合规 赔付费用
                    if (sumAkc254AddComplianceOwnExpense.doubleValue() <= 20000) {
                        sumAkc254AddCOECompensate = new BigDecimal("0");
                    } else {
                        for (double[] doubles : rate2) {
                            if (sumAkc254AddComplianceOwnExpense.doubleValue() > doubles[0] && sumAkc254AddComplianceOwnExpense.doubleValue() <= doubles[1]) {
//                                sumAkc254AddCOECompensate = (sumAkc254AddComplianceOwnExpense - doubles[0]) * doubles[2] + doubles[3];
                                sumAkc254AddCOECompensate = (sumAkc254AddComplianceOwnExpense.subtract(new BigDecimal(doubles[0]))).multiply(new BigDecimal(doubles[2])).add(new BigDecimal(doubles[3]));
                                break;
                            }
                        }
                    }

                } else {
                    if (sumAkc254.doubleValue() > 20000) {
                        sumAkc254Compensate = new BigDecimal("7750");
                    } else if (sumAkc254.doubleValue() < 3000) {
                        sumAkc254Compensate = new BigDecimal("0");
                    } else {
                        for (double[] doubles : rate3) {
                            if (sumAkc254.doubleValue() > doubles[0] && sumAkc254.doubleValue() <= doubles[1]) {
//                                sumAkc254Compensate = (sumAkc254 - doubles[0]) * doubles[2] + doubles[3];
                                sumAkc254Compensate = (sumAkc254.subtract(new BigDecimal(doubles[0]))).multiply(new BigDecimal(doubles[2])).add(new BigDecimal(doubles[3]));
                                break;
                            }
                        }
                    }

                    // 自负+ 计算后的合规 赔付费用
                    if (sumAkc254AddComplianceOwnExpense.doubleValue() <= 20000) {
                        sumAkc254AddCOECompensate = new BigDecimal("0");
                    } else {
                        for (double[] doubles : rate4) {
                            if (sumAkc254AddComplianceOwnExpense.doubleValue() > doubles[0] && sumAkc254AddComplianceOwnExpense.doubleValue() <= doubles[1]) {
//                                sumAkc254AddCOECompensate = (sumAkc254AddComplianceOwnExpense - doubles[0]) * doubles[2] + doubles[3];
                                sumAkc254AddCOECompensate = (sumAkc254AddComplianceOwnExpense.subtract(new BigDecimal(doubles[0]))).multiply(new BigDecimal(doubles[2])).add(new BigDecimal(doubles[3]));
                                break;
                            }
                        }
                    }
                }
                sumAkc254Compensate = sumAkc254Compensate.setScale(2, BigDecimal.ROUND_HALF_UP);
                sumAkc254AddCOECompensate = sumAkc254AddCOECompensate.setScale(2, BigDecimal.ROUND_HALF_UP);


                // 本次自负赔付
                BigDecimal subtract = sumAkc254Compensate.subtract(previousSumAkc254Compensate);
                thisData.setAkc254Compensate(subtract.toString());
                previousSumAkc254Compensate = sumAkc254Compensate;

                // 累计自负赔付
                thisData.setSumAkc254Compensate(sumAkc254Compensate.toString());
                // 累计自负费用
                thisData.setSumAkc254(sumAkc254.toString());
                // 累计合规费用
                thisData.setSumComplianceOwnExpense(sumComplianceOwnExpense.toString());
                // 累计自负费用+累计合规费用
                thisData.setSumAkc254AddComplianceOwnExpense(sumAkc254AddComplianceOwnExpense.toString());

                // 本次自负加合规赔付
                BigDecimal subtract1 = sumAkc254AddCOECompensate.subtract(previousSumAkc254AddCOECompensate);
                thisData.setAkc254AddCOECompensate(subtract1.toString());
                previousSumAkc254AddCOECompensate = sumAkc254AddCOECompensate;

                // 累计自负加合规赔付
                thisData.setSumAkc254AddCOECompensate(sumAkc254AddCOECompensate.toString());
                //获取系统当前时间
                SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
                String tCurrentDate = df.format(System.currentTimeMillis());
                thisData.setToday(tCurrentDate);
                context.write(thisData, NullWritable.get());

            }
        }

    }


    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        Job job = Job.getInstance(conf, CusJob.class.getSimpleName());
        job.setJarByClass(CusJob.class);

//        MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, CusJob.CulOldMapper.class);
//        MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, CusJob.CulMapper.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // map settings
        job.setMapperClass(CusJob.CulMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(CusBean.class);

        // reduce settings
        job.setReducerClass(CusJob.CulReduce.class);
        job.setOutputKeyClass(CusBean.class);
        job.setOutputValueClass(NullWritable.class);

        boolean res = job.waitForCompletion(true);

        return res ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", "hdfs://hadoop1:9000");
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        args = new String[]{"/dbyb/out/201803/t_db_intermediate_lastdata", "/dbyb/out/201803/t_db_calculation_results"};
        ToolRunner.run(configuration, new CusJob(), args);
    }
}
